package cn.young.dev.util;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

/**
 * @Auther: Ryan
 * @Date: 2022/1/21 14:05
 * @Description: 接收数据
 */
public class KafkaConsumer extends Thread{

    private final ConsumerConnector consumer;
    private final String topic;
    private final java.util.function.Consumer<String> retConsumer;

    public KafkaConsumer(String zookeeperConnect, String groupId, String topic, java.util.function.Consumer<String> retConsumer){
        this.topic = topic;
        this.retConsumer = retConsumer;
        Properties properties = new Properties();
        properties.put("zookeeper.connect", zookeeperConnect);
        properties.put("group.id", groupId);
        consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(properties));
    }

    @Override
    public void run() {
        Map<String, Integer> topicCountMap = new HashMap<>();
        topicCountMap.put(topic, 1);
        Map<String, List<KafkaStream<byte[], byte[]>>>  messageStreams = consumer.createMessageStreams(topicCountMap);
        KafkaStream<byte[], byte[]> stream = messageStreams.get(topic).get(0);
        for (kafka.message.MessageAndMetadata<byte[], byte[]> messageAndMetadata : stream) {
            String message = new String(messageAndMetadata.message());
            System.out.println("Receive: " + message);
            retConsumer.accept(message);
        }
    }
}
